package com.hujf.flink.tableSql;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import scala.Tuple2;

/**
 * @Author hujf
 * @Date 2022/8/26 10:35
 * @Description TODO
 */
public class CreateTable {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.setRuntimeMode(RuntimeExecutionMode.BATCH);

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
//
//        DataStreamSource<Tuple2> dataStreamSource = environment.fromElements(new Tuple2("Alice", 10),
//                new Tuple2<>("Bob", 12),
//                new Tuple2<>("Alice", 12));

        DataStream<Row> dataStreamSource = environment.fromElements( Row.of("Alice", 12),
                Row.of("Bob", 10),
                Row.of("Alice", 100));

        Table table = tableEnvironment.fromDataStream(dataStreamSource).as("name", "score");
        tableEnvironment.createTemporaryView("perScore",table);


        Table table1 = tableEnvironment.sqlQuery("select name, sum(score) as ct from perScore group by name");

        DataStream<Row> rowDataStream = tableEnvironment.toChangelogStream(table1);

        rowDataStream.print();

        environment.execute();

    }
}
